package cn.azzhu.day04;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author azzhu
 * @create 2020-09-19 17:12:01
 */
public class MySqlSink extends RichSinkFunction<ActivityBean> {
    private transient Connection connection = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?characterEncoding=utf-8","root","root");
    }

    @Override
    public void close() throws Exception {
       connection.close();
    }

    @Override
    public void invoke(ActivityBean bean, Context context) throws Exception {
        PreparedStatement pstm = null;
        try {
            pstm = connection.prepareStatement(
                    "insert  into t_activity_counts(aid,event_type,counts) values(?,?,?) on DUPLICATE KEY UPDATE counts = ?");
            pstm.setString(1, bean.aid);
            pstm.setInt(2,bean.eventType);
            pstm.setInt(3,bean.count);
            pstm.setInt(4,bean.count);
            //todo 无法保证exactly once
            pstm.executeUpdate();
        } finally {
           if(pstm != null) {
               pstm.close();
           }
        }
    }
}
